package com.consumer0.listener;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.stereotype.Component;

@Slf4j
@Component
public class SpringAmqpListener {

    public static final String QUEUE_INFORM_EMAIL = "queue_inform_email";
    public static final String QUEUE_INFORM_SMS = "queue_inform_sms";
    public static final String EXCHANGE_TOPICS_INFORM="exchange_topics_inform";
    public static final String ROUTINGKEY_EMAIL="inform.#.email.#";
    public static final String ROUTINGKEY_SMS="inform.#.sms.#";

//    @RabbitListener(queues = "dlx.queue")
//    public void listenDlxQueue2(String msg) throws InterruptedException {
//        log.info("消费者收到dlx.queue的消息" + msg);
//        Thread.sleep(200);
//    }

    @RabbitListener(bindings = @QueueBinding(
            value =@Queue(name = "topic.queue1",durable = "true"),
            exchange = @Exchange(name = "maxxub.topic",type = ExchangeTypes.TOPIC),
            key = {"#.new"}))
    public void listenTopicQueue1(String msg) throws InterruptedException {
        System.out.println("消费者1收到topic.queue1的消息");
        Thread.sleep(200);
    }
    @RabbitListener(bindings = @QueueBinding(
            value =@Queue(name = "topic.queue2",durable = "true"),
            exchange = @Exchange(name = "maxxub.topic",type = ExchangeTypes.TOPIC),
            key = {"china.#"}))
    public void listenTopicQueue2(String msg) throws InterruptedException {
        System.out.println("消费者2收到topic.queue2的消息"+msg);
        Thread.sleep(200);
    }

    @RabbitListener(bindings = @QueueBinding(
            value =@org.springframework.amqp.rabbit.annotation.Queue(name = "direct.queue1",durable = "true"),
            exchange = @Exchange(name = "maxxub.direct",type = ExchangeTypes.DIRECT),
            key = {"red","blue"}))
    public void listenDirectQueue1(String msg) throws InterruptedException {
        System.out.println("消费者1收到direct.queue1的消息"+msg);
        Thread.sleep(200);
    }
    @RabbitListener(bindings = @QueueBinding(
            value =@org.springframework.amqp.rabbit.annotation.Queue(name = "direct.queue2",durable = "true"),
            exchange = @Exchange(name = "maxxub.direct",type = ExchangeTypes.DIRECT),
            key = {"red","yellow"}))
    public void listenDirectQueue2(String msg) throws InterruptedException {
        System.out.println("消费者2收到direct.queue2的消息"+msg);
        Thread.sleep(200);
    }


    @RabbitListener(queues = "simple.queue")
    public void listenSimpleQueue1(String msg) throws InterruptedException {
        System.out.println("消费者1收到simple.queue的消息"+msg);
        Thread.sleep(2000);
        throw new InterruptedException("消费者1收到simple出现了异常！");
    }
    @RabbitListener(queues = "simple.queue")
    public void listenSimpleQueue2(String msg) throws InterruptedException {
        System.err.println("消费者2收到simple.queue的消息"+msg);
        Thread.sleep(1000);
    }


    @RabbitListener(queues = "simple.queue")   //使用该注解监听队列消息
    public void listenSimpleQueue0(String msg) throws InterruptedException {
        System.out.println("消费者0收到simple.queue的消息"+msg);
        Thread.sleep(2000);
    }

    //监听email队列
    @RabbitListener(queues = {QUEUE_INFORM_EMAIL})
    public void receive_email(Object msg, Message message, Channel channel){
        System.out.println("QUEUE_INFORM_EMAIL msg"+msg);
    }

    //监听sms队列
    @RabbitListener(queues = {QUEUE_INFORM_SMS})
    public void receive_sms(Object msg, Message message, Channel channel){
        System.out.println("QUEUE_INFORM_SMS msg"+msg);
    }

    /**
     * 用于死信交换机投递消息，当别的队列中消息处理错误被丢弃，或者超时未消费，则在该队列中进行处理
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "error.queue"),
            exchange = @Exchange(name = "test.exchange", type = ExchangeTypes.DIRECT),
            key = "error"
    ))
    public void listenerDeadQueue(String msg) {
        log.info("接收到 dead.queue 消息：" + msg);
        //在这里可以记录日志，或将消息存储到数据库中，后续由人工进行干预处理
    }

    //接收到消息后，处理程序会报异常
    //在 application.yml 中配置了本地重试 3 次，如果都是失败，默认会丢弃消息，返回 basic.reject
    //由于已经为该队列设置了死信交换机和相应的路由 key，因此最终失败的消息会投递到 error.queue 中
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "data.queue", arguments = {
                    //在队列上设置【死信交换机】和【路由key】
                    @Argument(name = "x-dead-letter-exchange", value = "test.exchange"),
                    @Argument(name = "x-dead-letter-routing-key", value = "error")}),
            exchange = @Exchange(name = "test.exchange", type = ExchangeTypes.DIRECT),
            key = "data"
    ))
    public void listenerDataQueue(String msg) {
        log.info("接收到 data.queue 消息：" + msg);
        //以下代码会抛出异常
        Integer result = 1 / 0;
    }

    //接收 header 设置了 x-delay 的延迟消息
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "test.queue"),
            //给 rabbitmq 安装了 delay message 插件后，给交换机设置 delayed 为 true，就变成了延迟交换机
            //当带有 x-delay 头的消息发送到该交换机后，消息先持久化到硬盘中，然后返回给生产者没有找到路由的错误异常，
            //因此生产者如果监听了消息是否由交换机发送到队列的话，就会收到 routing not found 的错误消息
            //此时生产者的回调方法中，需要判断返回的错误消息是否设置了 delay 以及 delay 值是否大于 0
            exchange = @Exchange(name = "delay.exchange", delayed = "true"),
            key = "delay"
    ))
    public void listenDelayExchange(String msg) {
        log.info("接收到了 test.queue 的延迟消息：" + msg);
    }


    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "lazy.queue",
                    //将队列设置为惰性队列，消息会持续稳定的存储到磁盘中，用于解决消息大量累积的问题
                    arguments = @Argument(name = "x-queue-mode", value = "lazy")),
            exchange = @Exchange(name = "test.exchange"),
            key = "lazy"
    ))
    public void listenLazyQueue(String msg) throws InterruptedException {
        log.info("接收到了 lazy.queue 的消息：" + msg);
        //模拟处理消息比较慢，造成消息无法及时消费处理掉，在队列中进行累积
        //如果采用惰性队列，则能够持续稳定的存储到硬盘中，防止消息丢失
        Thread.sleep(5000);
    }

}
